package cn.jly.flink.custom_source;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.ArrayList;
import java.util.Random;

/**
 * 自定义source
 *
 * @author lanyangji
 * @create 2020-09-01 15:14
 */
public class MyStreamingSource implements SourceFunction<MyStreamingSource.Item> {

    private boolean isRunning = true;

    /**
     * 重写run方法产生一个源源不断的数据发送源
     *
     * @param sourceContext
     * @throws Exception
     */
    @Override
    public void run(SourceContext<Item> sourceContext) throws Exception {
        while (isRunning) {
            sourceContext.collect(generateItem());
            
            // 每秒产生一条数据
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    /**
     * 随机产生一条商品数据
     *
     * @return
     */
    private Item generateItem() {
        Item item = new Item();

        ArrayList<String> list = new ArrayList<>();
        list.add("HAT");
        list.add("TIE");
        list.add("SHOE");

        item.setId(new Random().nextInt(10000));
        item.setName(list.get(new Random().nextInt(3)));
        return item;
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    @Builder
    public static class Item {
        private String name;
        private Integer id;
    }
}
